// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: daniel

#pragma once

#include <cstddef>
#include <string>

// interface
#include "coin2/exchange/base/order/order.h"
#include "coin2/strategy/strategy.h"

// message build
#include "coin/feed/fastfeed/feed.pb.h"
#include "pipeline/coin_request.pb.h"
#include "pipeline/coin_update.pb.h"

// feed
#include "coin2/exchange/base/feed/replay/proto_event.h"

namespace coin2::appcoin2::support {

namespace impl {

template <class T>
bool SerializeToString(const T& proto, std::string* data) {
  data->clear();
  if (!proto.SerializeToString(data)) {
    return false;
  }
  return true;
}

using coin::proto::OrderGatewayLog;
using coin::proto::OrderSpecProto;
using coin2::app::Driver;
using coin2::app::DriverConfig;
using coin2::exchange::base::feed::FeedUpdate;
using coin2::exchange::base::feed::proto::serialize::BookEventBuilder;
using coin2::exchange::base::feed::proto::serialize::FundingRateEventBuilder;
using coin2::exchange::base::feed::proto::serialize::IndexEventBuilder;
using coin2::exchange::base::feed::proto::serialize::IvGreeksEventBuilder;
using coin2::exchange::base::feed::proto::serialize::KlineEventBuilder;
using coin2::exchange::base::feed::proto::serialize::LiquidationEventBuilder;
using coin2::exchange::base::feed::proto::serialize::MarketAnalysisEventBuilder;
using coin2::exchange::base::feed::proto::serialize::MarkPriceEventBuilder;
using coin2::exchange::base::feed::proto::serialize::NavEventBuilder;
using coin2::exchange::base::feed::proto::serialize::OpenInterestEventBuilder;
using coin2::exchange::base::feed::proto::serialize::TradeEventBuilder;
using coin2::exchange::base::feed::proto::serialize::WalletEventBuilder;
using coin2::exchange::base::feed::timer::FeedTimer;
using coin2::exchange::base::order::BaseOrderSubsystem;
using coin2::exchange::base::order::OrderSpec;
using coin2::exchange::base::order::OrderUpdate;
using pipeline::AccountInfoProto;
using pipeline::CoinRequestProto;
using pipeline::CoinUpdateProto;
using pipeline::FeedUpdateProto;
using pipeline::OrderSnapshotProto;
using pipeline::OrderUpdateProto;
using pipeline::RiskUpdateProto;

class FeedUpdateConvertor {
 public:
  FeedUpdateConvertor() = default;

 public:
#define SERIALIZE_FEED_TO_STRING(ToProtoFunc)                  \
  auto feed_upd = PrepareProto(upd);                           \
  auto proto = ToProtoFunc(upd);                               \
  proto.set_mea(upd.mea().String());                           \
  proto.set_native_symbol(upd.product().feed_native_symbol()); \
  *feed_upd->mutable_fast_feed() = proto;                      \
  return SerializeToString(proto_, data);

  bool onBookFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(book_builder_.ToProto);
  }
  bool onTradeFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(TradeEventBuilder::ToProto);
  }
  bool onLiquidationFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(LiquidationEventBuilder::ToProto);
  }
  bool onOpenInterestFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(OpenInterestEventBuilder::ToProto);
  }
  bool onMarkPriceFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(MarkPriceEventBuilder::ToProto);
  }
  bool onIndexFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(IndexEventBuilder::ToProto);
  }
  bool onFundingRateFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(FundingRateEventBuilder::ToProto);
  }
  bool onIvGreeksFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(IvGreeksEventBuilder::ToProto);
  }
  bool onNavFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(NavEventBuilder::ToProto);
  }
  bool onKlineFeed(const FeedUpdate& upd, std::string* data) {
    const auto key = GetProductKey(upd);
    auto& kline_builder = kline_builder_map_[key];
    SERIALIZE_FEED_TO_STRING(kline_builder.ToProto);
  }
  bool onWalletFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(WalletEventBuilder::ToProto);
  }
  bool onMarketAnalysisFeed(const FeedUpdate& upd, std::string* data) {
    SERIALIZE_FEED_TO_STRING(MarketAnalysisEventBuilder::ToProto);
  }
  bool onHeartbeatFeed(const coin2::exchange::base::feed::timer::FeedTimer& timer) { return false; }

 private:
  auto PrepareProto(const FeedUpdate& upd) -> FeedUpdateProto* {
    proto_.Clear();
    proto_.set_type(CoinUpdateProto::FEED_UPDATE);
    auto* feed_upd = proto_.mutable_feed_update();
    feed_upd->set_type(FeedUpdateProto::FAST_FEED);
    auto* account_request = feed_upd->mutable_account_request();
    account_request->set_system_name(upd.system_name());
    account_request->set_market_type(coin::proto::MarketType_Name(upd.market()));
    account_request->set_exchange(coin::proto::ExchangeType_Name(upd.exchange()));
    account_request->set_api_version(upd.mea().api);
    return feed_upd;
  }

  auto GetProductKey(const FeedUpdate& upd) -> std::string {
    return fmt::format(
        "{}:{}:{}",
        MarketType_Name(upd.market()),
        ExchangeType_Name(upd.exchange()),
        upd.native_symbol());
  }

 private:
  CoinUpdateProto proto_;
  BookEventBuilder book_builder_{5};
  std::unordered_map<std::string, KlineEventBuilder> kline_builder_map_;
};

class OrderUpdateConvertor {
 public:
  OrderUpdateConvertor() = default;

 public:
  bool onAccountInfo(const OrderUpdate& upd, std::string*) {
    // TODO(daniel): it may be better to forward account info through interface onOrderLog
    return false;
  }
  bool onAccountOrder(const OrderUpdate& upd, std::string* data) {
    // TODO(daniel): it may be better to forward order update through interface onOrderLog
    return false;
  }
  bool onOrderLog(const OrderGatewayLog& og_log, std::string* data) {
    switch (og_log.type()) {
      case OrderGatewayLog::ORDER_EVENT:
        return HandleOrderEvent(og_log, data);
      case OrderGatewayLog::BALANCE:
        return HandleAccountBalance(og_log, data);
      case OrderGatewayLog::POSITION:
        return HandleAccountPosition(og_log, data);
      case OrderGatewayLog::TRANSFER:
        return false;
      case OrderGatewayLog::RISK_EVENT:
        return HandleRiskEvent(og_log, data);
      default:
        return false;
    }
  }

 private:
  bool HandleOrderEvent(const OrderGatewayLog& og_log, std::string* data) {
    proto_.Clear();
    proto_.set_type(CoinUpdateProto::ORDER_UPDATE);
    auto* order = proto_.mutable_order_update();
    order->set_type(OrderUpdateProto::ORDER_EVENT);
    auto* event = order->mutable_order_event();
    *event->mutable_account_request() = og_log.account_request();
    *event->mutable_order_event() = og_log.event();
    return SerializeToString(proto_, data);
  }

  bool HandleAccountBalance(const OrderGatewayLog& og_log, std::string* data) {
    proto_.Clear();
    proto_.set_type(CoinUpdateProto::ACCOUNT_INFO_UPDATE);
    auto* acc_info = proto_.mutable_account_info_update();
    acc_info->set_type(AccountInfoProto::BALANCE);
    *acc_info->mutable_account_request() = og_log.account_request();
    *acc_info->mutable_balance() = og_log.balance();
    return SerializeToString(proto_, data);
  }

  bool HandleAccountPosition(const OrderGatewayLog& og_log, std::string* data) {
    proto_.Clear();
    proto_.set_type(CoinUpdateProto::ACCOUNT_INFO_UPDATE);
    auto* acc_info = proto_.mutable_account_info_update();
    acc_info->set_type(AccountInfoProto::POSITION);
    *acc_info->mutable_account_request() = og_log.account_request();
    *acc_info->mutable_position() = og_log.position();
    return SerializeToString(proto_, data);
  }

  bool HandleRiskEvent(const OrderGatewayLog& og_log, std::string* data) {
    proto_.Clear();
    proto_.set_type(CoinUpdateProto::RISK_UPDATE);
    auto* risk = proto_.mutable_risk_update();
    risk->set_type(RiskUpdateProto::RISK_EVENT);
    *risk->mutable_account_request() = og_log.account_request();
    *risk->mutable_risk_event() = og_log.risk_event();
    return SerializeToString(proto_, data);
  }

 private:
  CoinUpdateProto proto_;
};

class OrderGatewayDelegate {
 public:
  explicit OrderGatewayDelegate(Driver* driver) : driver_{driver} {}

 public:
  bool CaptureOrderSnapshot(int64_t timestamp, std::string* data) {
    if (order_snapshot_last_sent_ts_ > 0 &&
        order_snapshot_last_sent_ts_ + order_snapshot_send_interval_ > timestamp) {
      return false;
    }
    order_snapshot_last_sent_ts_ = timestamp;
    const auto& systems = driver_->order()->GetAllSystems();
    if (systems.empty()) {
      return false;
    }
    // capture
    OrderSnapshotProto order_snapshot;
    for (auto& os : systems) {
      auto* order_context_bundle = order_snapshot.add_each_order_context_bundle();
      *order_context_bundle->mutable_account_request() = os->GetAccountRequest();
      for (auto* oc : os->gateway()->order_manager().GetAllWorkingOrders()) {
        *order_context_bundle->add_each_order_context() = oc->ToProto();
      }
    }
    // serialize
    upd_proto_.Clear();
    upd_proto_.set_type(CoinUpdateProto::ORDER_UPDATE);
    auto* order = upd_proto_.mutable_order_update();
    order->set_type(OrderUpdateProto::ORDER_SNAPSHOT);
    *order->mutable_order_snapshot() = order_snapshot;
    return SerializeToString(upd_proto_, data);
  }

  void HandleRequestMessage(const std::string& msg) {
    if (msg.empty()) {
      return;
    }

    try {
      req_proto_.Clear();
      req_proto_.ParseFromString(msg);  // TODO(daniel): should enable ignore unknown field ?
    } catch (std::exception& ex) {
      LOG(ERROR) << "invalid message: " << ex.what();
    }

    switch (req_proto_.type()) {
      case CoinRequestProto::SUBMIT_ORDER:
        HandleSubmitOrder(req_proto_);
        return;
      case CoinRequestProto::CANCEL_ORDER:
        HandleCancelOrder(req_proto_);
        return;
      case CoinRequestProto::AMEND_ORDER:
        HandleAmendOrder(req_proto_);
        break;
      default:
        LOG(ERROR) << "unsupported:\n" << req_proto_.DebugString();
        break;
    }
  }

  auto TryCancelAllWorkingOrders() -> int {
    int num_orders = 0;
    const auto& systems = driver_->order()->GetAllSystems();
    for (auto& os : systems) {
      auto working_orders(os->gateway()->order_manager().GetAllWorkingOrders());
      for (auto* oc : working_orders) {
        os->gateway()->CancelOrder(oc->proc_order_id());
        ++num_orders;
      }
    }
    return num_orders;
  }

  void onCleanupAux(int num_retries) {
    int num_orders = TryCancelAllWorkingOrders();
    if (num_orders == 0) {
      LOG(INFO) << "Cleared all pending orders.";
      driver_->PostTask([this]() { driver_->Exit(); });
    } else if (num_retries > 10) {
      LOG(WARNING) << "There may be some pendings orders left open.";
      driver_->PostTask([this]() { driver_->Exit(); });
    } else {
      driver_->PostDelayedTask(
          [num_retries, this]() { onCleanupAux(num_retries + 1); },
          500'000'000LL);
    }
  }

  void onCleanup() {
    driver_->SetImmediateExitOnSignal(false);
    driver_->PostTask([this]() { onCleanupAux(0); });
  }

 private:
  auto ToOrderSpec(BaseOrderSubsystem* os, OrderSpecProto* proto) -> std::unique_ptr<OrderSpec> {
    // TODO(daniel): is there any better solution?
    auto product = CreateProductFromNormStringNoThrow(
        proto->market_type(),
        proto->exchange_type(),
        proto->norm_product(),
        proto->order_created_time());
    if (product) {
      auto* product_holder = os->GetProductHolder(*product, proto->order_created_time());
      auto* tick_qty = &product_holder->product_info().tick_qty();
      auto order_qty = tick_qty->Floor(proto->qty());
      proto->set_qty(order_qty);
    }
    return OrderSpec::FromProto(*proto);
  }

  void HandleSubmitOrder(const CoinRequestProto& req_proto) {
    const auto& system_name = req_proto.account_request().system_name();
    auto* os = driver_->order()->GetUniqueSystemByName(system_name);
    if (nullptr == os) {
      LOG(ERROR) << fmt::format("failed to identify os: {}", system_name);
      return;
    }
    if (!os->is_ready()) {
      LOG(INFO) << "og is not ready";
      return;
    }
    auto* proto = const_cast<CoinRequestProto*>(&req_proto);
    auto order_spec = ToOrderSpec(os, proto->mutable_submit_order_spec());
    if (!order_spec) {
      LOG(ERROR) << fmt::format("failed to convert order spec: {}", req_proto.DebugString());
      return;
    }
    os->gateway()->SubmitOrder(*order_spec);
  }

  void HandleCancelOrder(const CoinRequestProto& req_proto) {
    const auto& system_name = req_proto.account_request().system_name();
    auto* os = driver_->order()->GetUniqueSystemByName(system_name);
    if (nullptr == os) {
      LOG(ERROR) << fmt::format("failed to identify os: {}", system_name);
      return;
    }
    if (!os->is_ready()) {
      LOG(INFO) << "og is not ready";
      return;
    }
    auto order_spec = req_proto.cancel_order_spec();
    os->gateway()->CancelOrder(order_spec.proc_order_id());
  }

  void HandleAmendOrder(const CoinRequestProto& req_proto) {
    LOG(INFO) << "@daniel: implement amending order\n" << req_proto.DebugString();
  }

 private:
  Driver* driver_;

  CoinUpdateProto upd_proto_;
  CoinRequestProto req_proto_;

  // periodic task
  int64_t order_snapshot_send_interval_ = 1'000'000'000LL;
  int64_t order_snapshot_last_sent_ts_ = 0;
};

template <class Strategy>
class ForwardingStrategy : public coin2::strategy::IStrategy {
 public:
  ForwardingStrategy() {
    static_assert(
        std::is_base_of<ForwardingStrategy, Strategy>::value,
        "ForwardingStrategy should be base class of Strategy");
    strategy_ = static_cast<Strategy*>(this);
  }
  ForwardingStrategy(const ForwardingStrategy&) = delete;
  ForwardingStrategy& operator=(const ForwardingStrategy&) = delete;
  virtual ~ForwardingStrategy() = default;

  /**************************************************************************************/
  /* implement interface                                                                */
  /**************************************************************************************/
 public:
  // feed
#define CONVERT_AND_SEND_FEED(ConvertFunc)                     \
  if (!feed_update_convertor_) {                               \
    return;                                                    \
  }                                                            \
  if (!feed_update_convertor_->ConvertFunc(upd, &send_buf_)) { \
    return;                                                    \
  }                                                            \
  Send(std::ref(send_buf_));

  void onBookFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onBookFeed);
  }
  void onTradeFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onTradeFeed);
  }
  void onLiquidationFeed(const FeedUpdate& upd) override {
    CONVERT_AND_SEND_FEED(onLiquidationFeed);
  }
  void onOpenInterestFeed(const FeedUpdate& upd) override {
    CONVERT_AND_SEND_FEED(onOpenInterestFeed);
  }
  void onMarkPriceFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onMarkPriceFeed);
  }
  void onIndexFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onIndexFeed);
  }
  void onFundingRateFeed(const FeedUpdate& upd) override {
    CONVERT_AND_SEND_FEED(onFundingRateFeed);
  }
  void onIvGreeksFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onIvGreeksFeed);
  }
  void onNavFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onNavFeed);
  }
  void onKlineFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onKlineFeed);
  }
  void onWalletFeed(const FeedUpdate& upd) override {
    // line break
    CONVERT_AND_SEND_FEED(onWalletFeed);
  }
  void onMarketAnalysisFeed(const FeedUpdate& upd) override {
    CONVERT_AND_SEND_FEED(onMarketAnalysisFeed);
  }
  void onHeartbeatFeed(const coin2::exchange::base::feed::timer::FeedTimer& timer) override {}

  // order
#define CONVERT_AND_SEND_ORDER(ConvertFunc)                        \
  if (!order_update_convertor_) {                                  \
    return;                                                        \
  }                                                                \
  if (!order_update_convertor_->ConvertFunc(og_log, &send_buf_)) { \
    return;                                                        \
  }                                                                \
  Send(std::ref(send_buf_));

  void onAccountInfo(const OrderUpdate& upd) override {}
  void onAccountOrder(const OrderUpdate& upd) override {}
  void onOrderLog(const OrderGatewayLog& og_log) override {
    // line break
    CONVERT_AND_SEND_ORDER(onOrderLog);
  }
  void onOrderSubsystemError(const OrderUpdate& upd) override {}

  // other
  void Init(::coin2::app::Driver* driver) override {
    // specific handler
    feed_update_convertor_ = std::make_unique<FeedUpdateConvertor>();
    order_update_convertor_ = std::make_unique<OrderUpdateConvertor>();
    order_gateway_delegate_ = std::make_unique<OrderGatewayDelegate>(driver);
    // init strategy
    strategy_->Start(driver);
  }
  void UpdateDriverConfig(coin2::app::DriverConfig* config) override {}
  void onCleanup(::coin2::app::Driver* driver) override {
    // stop strategy
    strategy_->Stop();
    // cleanup
    order_gateway_delegate_->onCleanup();
  }

  /**************************************************************************************/
  /* function                                                                           */
  /**************************************************************************************/
 public:
  void Poll() {
    CHECK(order_gateway_delegate_) << "handler is not initialized";
    while (Recv(&recv_buf_)) {
      order_gateway_delegate_->HandleRequestMessage(std::ref(recv_buf_));
      recv_buf_.clear();
    }
  }

  void Push() {
    auto ts = GetCurrentTimestamp();
    // add periodic task here
    if (order_gateway_delegate_->CaptureOrderSnapshot(ts, &send_buf_)) {
      Send(std::ref(send_buf_));
    }
  }

 private:
  template <class T>
  bool Send(T&& msg) {
    return strategy_->DoSend(std::forward<T>(msg));
  }

  template <class T>
  bool Recv(T* msg) {
    return strategy_->DoRecv(msg);
  }

 private:
  Strategy* strategy_;

  // message
  std::string send_buf_;
  std::string recv_buf_;

  // specific handler
  std::unique_ptr<FeedUpdateConvertor> feed_update_convertor_ = {};
  std::unique_ptr<OrderUpdateConvertor> order_update_convertor_ = {};
  std::unique_ptr<OrderGatewayDelegate> order_gateway_delegate_ = {};
};

}  // namespace impl

using impl::ForwardingStrategy;
}  // namespace coin2::appcoin2::support
